package com.lmind.core.message;

import java.net.InetSocketAddress;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import com.lmind.core.async.ResultHandler;
import com.lmind.core.gps.terminal.message.AccessMessage;

public class MessagePipeTestCase {
	
	private static ScheduledThreadPoolExecutor executor;
	
	private static MessagePipeIoHandler server = new MessagePipeIoHandler();
	
	private static MessagePipeIoHandler client = new MessagePipeIoHandler();
	
	private static NioSocketAcceptor acceptor;
	
	private static NioSocketConnector connector;
	
	private volatile static int port = 15443;
	
	private volatile static MessagePipe connectionAtServer;
	
	private volatile static MessagePipe connectionAtClient;
	
	private static Object serverReceivedMessage;
	
	private static Object clientReceivedMessage;
	
	private static MessageCodec codec = new JavaMessageCodec();

	@BeforeClass
	public static void breforeClass() throws Exception {
		
		executor = new ScheduledThreadPoolExecutor(5);
		
		MessagePipeIoHandlerConfig config = new MessagePipeIoHandlerConfig();
		config.setTimeout(10);
		
		acceptor = new NioSocketAcceptor();
        acceptor.setHandler(server);
        server.setIoService(acceptor);
        server.setCodec(codec);
        server.setConfig(config);
        server.addMessageServiceListener(new ServerListener());
        server.init();
        acceptor.bind(new InetSocketAddress(port));
	}
	
	@AfterClass
	public static void afterClass() throws Exception {
		acceptor.dispose(true);
		executor.shutdown();
	}
	
	@Before
	public void brefore() throws Exception {
		
		MessagePipeIoHandlerConfig config = new MessagePipeIoHandlerConfig();
		config.setKeepAliveTime(10);
        
        connector = new NioSocketConnector();
        connector.setHandler(client);
        client.setIoService(connector);
        client.setCodec(codec);
        client.setConfig(new MessagePipeIoHandlerConfig());
        client.init();
	}
	
	@After
	public void after() throws Exception {
		connector.dispose(true);
	}
	
	@Test
	public void test() throws Exception {
		
		if (!connector.connect(new InetSocketAddress("127.0.0.1", port)).await(1000)) {
			Assert.fail();
			return;
		}
		
		final Thread mainThread = Thread.currentThread();
		
        MessageServiceListener listener = new MessageServiceListener() {
			public void disconnected(MessagePipe pipe) {}
			public void connected(MessagePipe pipe) {}
			public void message(MessagePipe pipe, Object message) {
				Assert.assertTrue(message instanceof AccessMessage);
				AccessMessage msg = (AccessMessage)message;
				Assert.assertEquals("13112345678", msg.getDeviceSerial());
				mainThread.interrupt();
			}
			public void keepAlive(MessagePipe pipe) {}
		};
        server.addMessageServiceListener(listener);
		
		String sim = "13112345678";
		AccessMessage message = new AccessMessage(sim);
		client.getMessagePipes().get(0).send(message);

		boolean b = false;
		try {
			Thread.sleep(5000);
		} catch (InterruptedException e) {
			b = true;
		}
		Assert.assertTrue(b);
		server.removeMessageServiceListener(listener);
	}
	
	@Test
	public void test2() throws Exception {
		
		if (!connector.connect(new InetSocketAddress("127.0.0.1", port)).await(1000)) {
			Assert.fail();
			return;
		}
		
		final int loop = 1000;
		String sim = "13112345678";
		AccessMessage message = new AccessMessage(sim);
		final Thread mainThread = Thread.currentThread();
		
        class Listener implements MessageServiceListener {
        	private AtomicInteger count = new AtomicInteger(0);
			public void disconnected(MessagePipe pipe) {}
			public void connected(MessagePipe pipe) {}
			public void message(MessagePipe pipe, Object message) {
				if (count.incrementAndGet() == loop) {
					mainThread.interrupt();
				}
			}
			public void keepAlive(MessagePipe pipe) {}
		};
		Listener listener = new Listener();
        server.addMessageServiceListener(listener);
		
        MessagePipe pipe = client.getMessagePipes().get(0);
		long start = System.currentTimeMillis();
		for (int i = 0; i < loop; i++) {
			pipe.send(message);
		}
		
		boolean timeout = true;
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			timeout = false;
		}
		long end = System.currentTimeMillis();
		System.out.println("time = " + (end - start));
		
		Assert.assertTrue(!timeout);
		Assert.assertEquals(listener.count.get(), loop);
		
		server.removeMessageServiceListener(listener);
	}
	
	static class ServerListener implements MessageServiceListener {
		public void connected(MessagePipe connection) {
			connectionAtServer = connection;
		}

		public void disconnected(MessagePipe connection) {
		}

		public void message(MessagePipe MessagePipe, Object message) {
			serverReceivedMessage = message;
		}

		@Override
		public void keepAlive(MessagePipe connection) {
			// TODO Auto-generated method stub
			
		}
	}
	
	static class ClientListener implements MessageServiceListener {
		public void connected(MessagePipe connection) {
			connectionAtClient = connection;
		}

		public void disconnected(MessagePipe connection) {
		}

		public void message(MessagePipe connection, Object message) {
			clientReceivedMessage = message;
		}

		@Override
		public void keepAlive(MessagePipe connection) {
			// TODO Auto-generated method stub
			
		}
	}

}
